package com.mjf.loginfail_detect

import java.net.URL
import java.util

import com.mjf.dim.{LoginEvent, Warning}
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor, ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

/**
 * 监测连续两次登录失败的用户
 *
 * 使用状态编程实现程序会有问题：
 *  比如 必须等到2s才会触发定时器
 *  比如 乱序数据处理
 */
object LoginFail {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 从文件读取数据并处理
    val resource: URL = getClass.getResource("/LoginLog.csv")
    val loginEventStream: DataStream[LoginEvent] = env.readTextFile(resource.getPath)
      .map {
        line =>
          val dataArr: Array[String] = line.split(",")
          LoginEvent(dataArr(0).toLong, dataArr(1), dataArr(2), dataArr(3).toLong)
      }.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LoginEvent](Time.seconds(3)) {
      override def extractTimestamp(element: LoginEvent): Long = element.eventTime * 1000L
    })

    // 用ProcessFunction进行转换，如果遇到2秒内连续两次登录失败，就输出报警
    val resultStream: DataStream[Warning] = loginEventStream
      .keyBy(_.userId)
      .process(new LoginFailWarning(2))

    resultStream.print()

    env.execute("LoginFail")

  }
}

class LoginFailWarning(maxFailTimes: Int) extends KeyedProcessFunction[Long, LoginEvent, Warning] {

  // 定义List状态，用来保存2秒内所有的登录失败事件
  lazy val loginFailListState: ListState[LoginEvent] = getRuntimeContext.getListState(new ListStateDescriptor[LoginEvent]("saved-loginfail", classOf[LoginEvent]))

  // 定义Value状态，用来保存定时器的时间戳
  lazy val timerTsState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("timer-ts", classOf[Long]))

  override def processElement(value: LoginEvent, ctx: KeyedProcessFunction[Long, LoginEvent, Warning]#Context, out: Collector[Warning]): Unit = {

    // 判断当前数据是否为登录失败
    if(value.eventType == "fail") {
      // 如果是登录失败，那么添加到ListState里，如果没有注册过定时器，就注册
      loginFailListState.add(value)
      if(timerTsState.value() == 0) {
        val ts: Long = value.eventTime * 1000L + 2000L
        ctx.timerService().registerEventTimeTimer(ts)
        timerTsState.update(ts)
      }
    } else {
      // 如果是登录成功，删除定时器，重新开始
      ctx.timerService().deleteEventTimeTimer(timerTsState.value())
      loginFailListState.clear()
      timerTsState.clear()
    }

  }

  // 如果两秒后的定时器触发了，那么判断ListState中失败的个数
  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, LoginEvent, Warning]#OnTimerContext, out: Collector[Warning]): Unit = {
    val allLoginFailList: ListBuffer[LoginEvent] = new ListBuffer[LoginEvent]

    val iter: util.Iterator[LoginEvent] = loginFailListState.get().iterator()

    while(iter.hasNext) {
      allLoginFailList += iter.next()
    }

    if(allLoginFailList.size >= maxFailTimes) {
      out.collect(Warning(
        ctx.getCurrentKey,
        allLoginFailList.head.eventTime,
        allLoginFailList.last.eventTime,
        "login fail in 2s for " + allLoginFailList.size + " times"
      ))
    }

    loginFailListState.clear()
    timerTsState.clear()

  }
}
